package com.iphanta.community.mq;

import com.iphanta.community.activity.photographic.common.BallBox;
import com.iphanta.community.activity.photographic.common.BallBoxCollection;
import com.iphanta.community.activity.photographic.entity.ActivityEntry;
import com.iphanta.community.activity.photographic.mapper.ActivityEntryMapper;
import com.iphanta.community.cache.ActivityCache;
import com.iphanta.community.common.utils.RedisKey;
import com.iphanta.community.common.utils.RedisKeyUtils;
import com.iphanta.community.mq.message.ActivityPost;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.redisson.api.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.Objects;
import java.util.Set;

/**
 * @author 墨声 E-mail: zehong.hongframe.huang@gmail.com
 * create time: 4/13/22 11:17 AM
 */
//@Service
//@RocketMQMessageListener(
//        topic = "photographic-post",
//        consumerGroup = "photographic-group",
//        consumeThreadMax = 32
//)
public class PhotographicPostListener implements RocketMQListener<ActivityPost> {

    private final static Logger LOGGER = LoggerFactory.getLogger(PhotographicPostListener.class);

    //    @Autowired
    private ActivityEntryMapper activityEntryMapper;

    @Autowired
    private RedissonClient redissonClient;

    @Autowired
    private ActivityCache activityCache;

    @Override
    public void onMessage(ActivityPost message) {
        MDC.put("requestId", message.getRequestId());

        String entrySeqKey = RedisKeyUtils.INSTANCE.key(RedisKey.ATOMIC_LONG_ACTIVITY_ENTRY_SEQ, message.getActivityId());
        RAtomicLong entrySeq = this.redissonClient.getAtomicLong(entrySeqKey);
        if (entrySeq.get() > 10 && Objects.isNull(this.activityCache.get(message.getActivityId()))) {
            this.activityCache.put(message.getActivityId(),
                    new BallBoxCollection(redissonClient, message.getActivityId()));
        }

        ActivityEntry activityEntry = new ActivityEntry();
        activityEntry.setActivityId(message.getActivityId());
        activityEntry.setDynamicId(message.getDynamicId());
        activityEntry.setEntryId(entrySeq.incrementAndGet());
//   TODO     this.activityEntryMapper.insert(activityEntry);

        RTransaction transaction = this.redissonClient.createTransaction(TransactionOptions.defaults().retryAttempts(3));
        try {
            RSet<Long> entryIds = transaction.getSet(RedisKeyUtils.INSTANCE.key("activity-post", message.getActivityId()));
            entryIds.add(activityEntry.getEntryId());

            RBucket<String> dynamicIdBucket = transaction.getBucket(RedisKeyUtils.INSTANCE.key("activity-post", "dynamicId", String.valueOf(activityEntry.getEntryId())));
            dynamicIdBucket.set(String.valueOf(activityEntry.getDynamicId()));

            String entryKey = RedisKeyUtils.INSTANCE.key("bucket_ball_box", String.valueOf(message.getDynamicId()));
            RBucket<BallBox> bucket = transaction.getBucket(entryKey);
            //TODO 读取活动配置
            BallBox box = new BallBox(message.getDynamicId(), 20, 20);
            box.setActivityId(message.getActivityId());
            box.setEntryId(activityEntry.getEntryId());
            bucket.set(box);

            transaction.commit();
        } catch (Exception e) {
            LOGGER.error("", e);
            transaction.rollback();
        } finally {
            MDC.remove("requestId");
        }

    }
}
